package com.ayu.flink.processfunction;

import com.ayu.flink.datasteamapibase.ClickSource;
import com.ayu.flink.datasteamapibase.Event;
import com.ayu.flink.timelineandwindow.UrlViewCountPoJo;
import com.ayu.flink.timelineandwindow.UvCountExample;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Author: 徐林玉//作者及
 * @Date: 2022/3/24//完成日期
 * @Description: //
 * @Version: v0.0.1 // 版本信息
 * @Function List: // 主要函数及其功能
 * @Others: // 其它内容的说明
 * @History: // 历史修改记录
 **/
public class KeyedProcessTopNExample
{

	public static void main(String[] args) throws Exception
	{

		//执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		//水位线时间戳
		SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource()).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps().withTimestampAssigner((SerializableTimestampAssigner<Event>) (event, recordTimestamp) -> event.timestamp));

		//开窗 10秒内的结果 5秒更新一次
		SingleOutputStreamOperator<UrlViewCountPoJo> urlCountStream = eventStream.keyBy(data -> data.url).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))).aggregate(new UvCountExample.UrlCountAgg(), new UvCountExample.UrlCountResult());

		urlCountStream.print("source");
		// 对结果中同一个窗口的统计数据，进行排序处理
		SingleOutputStreamOperator result = urlCountStream.keyBy(UrlViewCountPoJo::getWindowEnd).process(new TopN(5));

		result.print("result");

		env.execute();

	}

	public static class TopN extends KeyedProcessFunction<Long, UrlViewCountPoJo, String>
	{

		private Integer n;

		private ListState<UrlViewCountPoJo> listState;

		public TopN(Integer n)
		{
			this.n = n;
		}

		@Override
		public void open(Configuration parameters) throws Exception
		{
			// 从环境中获取列表状态句柄
			listState = getRuntimeContext().getListState(new ListStateDescriptor<UrlViewCountPoJo>("yurl-viewcount-list", Types.POJO(UrlViewCountPoJo.class)));

		}

		@Override
		public void processElement(UrlViewCountPoJo value, Context ctx, Collector<String> out) throws Exception
		{
			// 将count数据添加到列表状态中，保存起来
			listState.add(value);
			// 注册 window end + 1ms后的定时器，等待所有数据到齐开始排序
			ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);

		}

		@Override
		public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception
		{

			// 将数据从列表状态变量中取出，放入ArrayList，方便排序
			List<UrlViewCountPoJo> resList = new ArrayList<>();

			for (UrlViewCountPoJo viewCount : listState.get())
			{
				resList.add(viewCount);
			}
			// 清空状态，释放资源
			listState.clear();
			resList = resList.stream().sorted(Comparator.comparingLong(UrlViewCountPoJo::getCount).reversed()).collect(Collectors.toList());

			//构建输出结果
			StringBuilder result = new StringBuilder();
			result.append("========================================\n");
			result.append("窗口结束时间：" + new Timestamp(timestamp - 1) + "\n");
			for (int i = 0; i < n; i++)
			{

				try
				{

					UrlViewCountPoJo res = resList.get(i);
					String info = "No." + (i + 1) + " " + "url：" + res.getUrl() + " " + "浏览量：" + res.getCount() + "\n";
					result.append(info);

				} catch (Exception e)
				{

				}

			}
			result.append("========================================\n");
			out.collect(result.toString());
		}
	}
}
